package org.hong.monkey.util

import java.util.logging.Level

import akka.actor.{ActorRef, ExtendedActorSystem, ActorSystem}
import akka.pattern.ask
import com.typesafe.config.ConfigFactory
import org.apache.log4j.Level._
import org.apache.log4j.Logger
import org.hong.monkey.rpc.RpcTimeout
import org.hong.monkey.{MonkeyException, MonkeyEnv, MonkeyConf, Logging}
import scala.collection.JavaConverters._

private[monkey] object AkkaUtils extends Logging {

  def createActorSystem(name:String,
                        host:String,
                        port:Int,
                        conf:MonkeyConf): (ActorSystem, Int) = {
    val startService:Int => (ActorSystem, Int) = {
      actualPort => doCreateActorSystem(name, host, actualPort, conf)
    }
    Utils.startServiceOnPort(port, startService, conf, name)
  }

  private def doCreateActorSystem(name:String,
                                   host:String,
                                   port:Int,
                                   conf:MonkeyConf): (ActorSystem, Int) = {
    val akkaThreads = conf.getInt("monkey.akka.threads", 4)
    val akkaBatchSize = conf.getInt("monkey.akka.batchSize", 15)
    val akkaTimeoutS = conf.getTimeAsSeconds("monkey.akka.timeout", conf.get("monkey.network.timeout", "120s"))
    val akkaFrameSize = maxFrameSizeBytes(conf)
    val akkaLogLifecycleEvents = conf.getBoolean("monkey.akka.logLifecycleEvents", false)
    val lifecycleEvents = if (akkaLogLifecycleEvents) "on" else "off"
    if (!akkaLogLifecycleEvents) {
      Option(Logger.getLogger("akka.remote.EndpointWriter")).map(l => l.setLevel(FATAL))
    }

    val logAkkaConfig = if (conf.getBoolean("monkey.akka.logAkkaConfig", false)) "on" else "off"

    val akkaHeartBeatPausesS = conf.getTimeAsSeconds("monkey.akka.hartbeat.pauses", "6000s")
    val akkaHeartBeatIntervalS = conf.getTimeAsSeconds("monkey.akka.hartbeat.interval", "1000s")

    val akkaConf = ConfigFactory.parseMap(conf.getAkkaConf.toMap.asJava)
    .withFallback(ConfigFactory.parseString(
    s"""
       |akka.daemonic = no
       |akka.loggers = [""akka.event.slf4j.Slf4jLogger""]
       |akka.stdout-loglevel = "ERROR"
       |akka.jvm-exit-on-fatal-error = off
       |akka.remote.require-cookie = "no"
       |akka.remote.secure-cookie = ""
       |akka.remote.transport-failure-detector.hartbeat-interval = $akkaHeartBeatIntervalS
       |akka.remote.transport-failure-detector.acceptable.hartbeat-pause = $akkaHeartBeatPausesS
       |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.transport-class = "akka.remote.transport.netty.NettyTransport"
       |akka.remote.netty.tcp.hostname = "$host"
       |akka.remote.netty.tcp.port = "$port"
       |akka.remote.netty.tcp.tcp-nodelay = "no"
       |akka.remote.netty.tcp.connection-timeout = $akkaTimeoutS s
       |akka.remote.netty.tcp.maximum-frame-size = ${akkaFrameSize} B
       |akka.remote.netty.tcp.execution-pool-size = $akkaThreads
       |akka.actor.deault-dispatcher.throughput = $akkaBatchSize
       |akka.log-config-on-start = $logAkkaConfig
       |akka.remote.log-remote-lifecycle-events = $lifecycleEvents
       |akka.log-dead-letters = $lifecycleEvents
       |akka.log-dead-letters-during-shutdown = $lifecycleEvents
     """.stripMargin))

    val actorSystem = ActorSystem(name, akkaConf)
    val provider = actorSystem.asInstanceOf[ExtendedActorSystem].provider
    val boundPort = provider.getDefaultAddress.port.get
    (actorSystem, boundPort)
  }

  private val AKKA_MAX_FRAME_SIZE_IN_MB = Int.MaxValue / 1024 / 1024

  def maxFrameSizeBytes(monkeyConf: MonkeyConf): Int = {
    val frameSizeInMB = monkeyConf.getInt("monkey.akka.frameSize", 128)
    if (frameSizeInMB > AKKA_MAX_FRAME_SIZE_IN_MB)
      throw new IllegalArgumentException(s"monkey.akka.frameSize should not be greater than $AKKA_MAX_FRAME_SIZE_IN_MB MB")
    frameSizeInMB * 1024 * 1024
  }

  val reservedSizeBytes = 200 * 1024

  def askWithReply[T](
                       message: Any,
                       actor: ActorRef,
                       timeout: RpcTimeout): T = {
    askWithReply[T](message, actor, maxAttempts = 1, retryInterval = Int.MaxValue, timeout)
  }

  def askWithReply[T] (
    message: Any,
    actor: ActorRef,
    maxAttempts: Int,
    retryInterval: Long,
    timeout: RpcTimeout): T = {
    // TODO: Consider removing multiple attempts
    if (actor == null) {
      throw new MonkeyException(s"Error sending message [message = $message]" +
        " as actor is null ")
    }
    var attempts = 0
    var lastException: Exception = null
    while (attempts < maxAttempts) {
      attempts += 1
      try {
        val future = actor.ask(message)(timeout.duration)
        val result = timeout.awaitResult(future)
        if (result == null) {
          throw new MonkeyException("Actor returned null")
        }
        return result.asInstanceOf[T]
      } catch {
        case ie: InterruptedException => throw ie
        case e: Exception =>
          lastException = e
          logWarning(s"Error sending message [message = $message] in $attempts attempts", e)
      }
      if (attempts < maxAttempts) {
        Thread.sleep(retryInterval)
      }
    }

    throw new MonkeyException(
      s"Error sending message [message = $message]", lastException)
  }

  def makeDriverRef(name: String, conf: MonkeyConf, actorSystem: ActorSystem): ActorRef = {
    val driverActorSystemName = MonkeyEnv.driverActorSystemName
    val driverHost: String = conf.get("monkey.driver.host", "localhost")
    val driverPort: Int = conf.getInt("monkey.driver.port", 7077)
    Utils.checkHost(driverHost, "Expected hostname")
    val url = address(protocol(actorSystem), driverActorSystemName, driverHost, driverPort, name)
    val timeout = RpcUtils.lookupRpcTimeout(conf)
    logInfo(s"Connecting to $name: $url")
    timeout.awaitResult(actorSystem.actorSelection(url).resolveOne(timeout.duration))
  }

  def makeExecutorRef(
                       name: String,
                       conf: MonkeyConf,
                       host: String,
                       port: Int,
                       actorSystem: ActorSystem): ActorRef = {
    val executorActorSystemName = MonkeyEnv.executorActorSystemName
    Utils.checkHost(host, "Expected hostname")
    val url = address(protocol(actorSystem), executorActorSystemName, host, port, name)
    val timeout = RpcUtils.lookupRpcTimeout(conf)
    logInfo(s"Connecting to $name: $url")
    timeout.awaitResult(actorSystem.actorSelection(url).resolveOne(timeout.duration))
  }

  def protocol(actorSystem: ActorSystem): String = {
    val akkaConf = actorSystem.settings.config
    val sslProp = "akka.remote.netty.tcp.enable-ssl"
    protocol(akkaConf.hasPath(sslProp) && akkaConf.getBoolean(sslProp))
  }

  def protocol(ssl: Boolean = false): String = {
    if (ssl) {
      "akka.ssl.tcp"
    } else {
      "akka.tcp"
    }
  }

  def address(
    protocol: String,
    systemName: String,
    host: String,
    port: Int,
    actorName: String): String = {
    s"$protocol://$systemName@$host:$port/user/$actorName"
  }
}
